package com.deodar.kettle.platform.monitor.service;

import com.deodar.kettle.platform.common.App;
import com.deodar.kettle.platform.common.util.KettleConstant;
import com.deodar.kettle.platform.common.util.RepositoryUtils;
import com.deodar.kettle.platform.database.domain.RTransformation;
import com.deodar.kettle.platform.database.service.IRDirectoryService;
import com.deodar.kettle.platform.database.service.IRTransformationService;
import com.deodar.kettle.platform.monitor.executor.TransExecutor;
import com.deodar.kettle.platform.monitor.util.KettlePoolUtil;
import com.deodar.kettle.platform.monitor.util.TaskManageUtil;
import com.deodar.kettle.platform.parameter.service.IRnParameterService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.pentaho.di.base.AbstractMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.repository.Repository;
import org.pentaho.di.trans.TransExecutionConfiguration;
import org.pentaho.di.trans.TransMeta;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

import java.util.Map;

/**
 * @author: administrator
 * @Description:
 * @version:1.0.0
 */
@Slf4j
@Service("transService")
public class TransService {

    @Autowired
    @Qualifier("transformationService")
    IRTransformationService transformationService;

    @Autowired
    @Qualifier("directoryService")
    IRDirectoryService directoryService;

    @Autowired
    IRnParameterService bnParameterService;

    public boolean delete(Integer transId){
        Repository repository = App.getInstance().getRepository();
        try {
            repository.deleteTransformation(RepositoryUtils.getObjectId(transId));
            return true;
        } catch (KettleException e) {
            log.error(ExceptionUtils.getStackTrace(e));
        }
        return false;
    }

    public boolean start(Integer transId){
        return startTrans(transId,null);
    }

    public boolean startTrans(Integer transId,String logId){

        RTransformation transformation = transformationService.selectRTransformationById(transId);
        try {
            Repository repository = App.getInstance().getRepository();
            TransMeta transMeta = repository.loadTransformation(RepositoryUtils.getObjectId(transId),null) ;
                    //RepositoryUtils.loadTransByPath(path);
            TransExecutor transExecutor = new TransExecutor(decode(null,bnParameterService.selectMapByGroupId(transformation.getParGroupId())),transMeta,logId);
            KettlePoolUtil.getInstance().run(transExecutor);
            return true;
        } catch (KettleException e) {
            log.error(ExceptionUtils.getStackTrace(e));
        }
        return false;
    }

    public TransExecutionConfiguration decode(AbstractMeta meta, Map<String,String> mapVariables){
        TransExecutionConfiguration transExecutionConfiguration = new TransExecutionConfiguration();
        if(mapVariables != null) {
            transExecutionConfiguration.setVariables(mapVariables);
        }
        transExecutionConfiguration.setExecutingRemotely(true);
        transExecutionConfiguration.setRepository(App.getInstance().getRepository());
        return  transExecutionConfiguration;

    }


    public boolean stop(String logId){
        String key = KettleConstant.KEY_PREFIX+logId;
        TransExecutor transExecutor = (TransExecutor) TaskManageUtil.getInstance().get(key);
        if(transExecutor != null){
            transExecutor.stop();
            transExecutor.setClickStop(true);
            log.info("接收停止信号，开始停止trans任务");
            return true;
        }
        return false;
    }
}
